package com.test.reactive;


import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

/**
 * Created by jl on 2019/1/28 0028
 */
public class Demo {
    public static void main1(String[] args) {
        Flux<String> flux = Flux.just("red", "white", "blue");
        flux.map(String::toUpperCase).log().subscribe(new Subscriber<String>() {
            private long count = 0;
            private Subscription subscription;

            @Override
            public void onSubscribe(Subscription subscription) {
                this.subscription = subscription;
                subscription.request(2);
            }

            @Override
            public void onNext(String s) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                count++;
                if (count >= 2) {
                    count = 0;
                    subscription.request(2);
                }
                System.out.println(s);
            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onComplete() {
                System.out.println("complete...");
            }
        });

        // 最后输出该句，可见未开启新的线程
        System.out.println("end of this thread");
    }

    public static void main(String[] args) {
        Flux.just("red", "white", "blue")
                .log()
                .flatMap(value ->
                        Mono.just(value.toUpperCase()).subscribeOn(Schedulers.parallel()), 2)
                .subscribe(value->{
                    System.out.println(Thread.currentThread().getName() + " consumed: " + value);
                });
    }
}
